package com.dongdongshop.mq;

import com.alibaba.fastjson.JSON;
import com.dongdongshop.pojo.TbUserScoreAccount;
import com.dongdongshop.service.OrderService;
import com.dongdongshop.service.ScoreAccountService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "order-score-consumer-group",topic = "order-score-topic")
public class MQConsumer implements RocketMQListener<String> , RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @DubboReference
    private OrderService orderService;
    @DubboReference
    private ScoreAccountService scoreAccountService;

    //没有ACK 了， 也就是说这种方式不适合生产环境使用
    //一旦重写了prepareStart方法，此方法失效，但是必须实现RocketMQListener接口
    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                MessageExt messageExt = list.get(0);
                System.out.println("消息主题:"+messageExt.getTopic());
                String msgId = messageExt.getMsgId();
                System.out.println("消息ID:"+messageExt.getMsgId());
                System.out.println("消息重试次数:"+messageExt.getReconsumeTimes());
                System.out.println("消息队列ID:"+messageExt.getQueueId());
                byte[] body = messageExt.getBody();

                try {
                    String msg = new String(body,"utf-8");
                    //先判断有没有消费过，如果没消费过再消费，如果消费过，就不再重复消费
                    if (redisTemplate.hasKey(msgId)){
                        System.out.println("表示这条消息已经消费了");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    TbUserScoreAccount scoreAccount = JSON.parseObject(msg,TbUserScoreAccount.class);
                    Boolean b = scoreAccountService.insertInfo(scoreAccount);
                    System.out.println("消息:"+scoreAccount);
                    redisTemplate.boundValueOps(msgId).set("1");
                }catch (Exception e){
                    e.printStackTrace();
                    if (messageExt.getReconsumeTimes() < 3){
                         return ConsumeConcurrentlyStatus.RECONSUME_LATER;//稍后再试
                    }
                    //写入死信队列
                    System.out.println("由于消息失败次数过多,加入死信队列");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//消费成功
            }
        });
    }
}
